#include <brpc/server.h>
#include <butil/logging.h>

#include "etcd.hpp"     // 服务注册模块封装
#include "logger.hpp"   // 日志模块封装
#include "rabbitmq.hpp"
#include "channel.hpp"
#include "utils.hpp"
#include "mysql_chat_session_member.hpp"

#include "base.pb.h"  // protobuf框架代码
#include "user.pb.h"  // protobuf框架代码
#include "transmite.pb.h"  // protobuf框架代码

class MsgTransmitServiceImpl : public lch_im::MsgTransmitService
{
    public:
    MsgTransmitServiceImpl(const std::string &user_service_name,
            const ServiceManager::ptr &channels,
            const std::shared_ptr<odb::core::database> &mysql_client,
            const std::string &exchange_name,
            const std::string &routing_key,
            const MQClient::ptr &mq_client):
            _user_service_name(user_service_name),
            _manager_channels(channels),
            _mysql_session_member_table(std::make_shared<ChatSessionMemberTable>(mysql_client)),
            _exchange_name(exchange_name),
            _routing_key(routing_key),
            _mq_client(mq_client){}

    ~MsgTransmitServiceImpl() {}

    void GetTransmitTarget(google::protobuf::RpcController *controller,
                            const lch_im::NewMessageReq *request,
                            lch_im::GetTransmitTargetRsp *response,
                            ::google::protobuf::Closure *done)
    {
        LOG_DEBUG("收到消息转发请求");
        brpc::ClosureGuard rpc_guard(done);
        auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(errmsg);
        };

        // 1. 从请求中取出消息内容，会话 ID， 用户 ID
        std::string rid = request->request_id();
        std::string uid = request->user_id();
        std::string chat_ssid = request->chat_session_id();
        const lch_im::MessageContent& content = request->message(); 
        //进行消息组织，知道发送者，用户子服务获取信息  例如 所属的会话ID，消息内容
        auto channel = _manager_channels->choose(_user_service_name);
        if(!channel) {
            LOG_ERROR("{}-{} 没有可提供访问的用户子服务节点",rid,_user_service_name);
            return err_response(rid,"没有可提供访问的用户子服务节点");
        }

      
        // 2. 根据用户 ID 从用户子服务获取当前发送者用户信息
        lch_im::UserService_Stub stub(channel.get());
        lch_im::GetUserInfoReq req;
        lch_im::GetUserInfoRsp rsp;
        req.set_request_id(rid);
        req.set_user_id(uid);
        brpc::Controller cntl;
        stub.GetUserInfo(&cntl, &req, &rsp, nullptr);
        if (cntl.Failed() == true || rsp.success() == false)
        {
            LOG_ERROR("{} 用户子服务调用失败: {}", uid, cntl.ErrorText());
            return err_response(request->request_id(), "用户子服务调用失败!");
        }

        // 3. 根据消息内容构造完成的消息结构（分配消息 ID，填充发送者信息，填充消息产生时间）
        lch_im::MessageInfo message;
        message.set_message_id(uid);
        message.set_chat_session_id(chat_ssid);
        message.set_timestamp(time(nullptr));
        message.mutable_sender()->CopyFrom(rsp.user_info());
        message.mutable_message()->CopyFrom(content);
        auto target_list = _mysql_session_member_table->members(chat_ssid);
        // 4. 将消息序列化后发布到 MQ 消息队列中，让消息存储子服务对消息进行持久化存储
        bool ret = _mq_client->publish(_exchange_name, message.SerializeAsString(), _routing_key);
        if (ret == false) {
            LOG_ERROR("{} - 持久化消息发布失败：{}！", request->request_id(), cntl.ErrorText());
            return err_response(request->request_id(), "持久化消息发布失败：!");
        }
        // 5. 从数据库获取目标会话所有成员 ID
        response->set_request_id(rid);
        response->set_success(true);
        response->mutable_message()->CopyFrom(message);
        for(auto id : target_list) {
            response->add_target_id_list(id);
        } 
        LOG_DEBUG("处理消息转发请求完毕");
    }

private:
    //用户子服务
    std::string _user_service_name;
    ServiceManager::ptr _manager_channels;

    ChatSessionMemberTable::ptr _mysql_session_member_table;

    //消息队列客户端句柄
    std::string _exchange_name;
    std::string _routing_key;
    MQClient::ptr _mq_client;
};

class TransmiteServer
{
public:
    using ptr = std::shared_ptr<TransmiteServer>;
    TransmiteServer(const std::shared_ptr<odb::core::database> &mysql_client,
                    const Discovery::ptr discovery_client,
                    const Registery::ptr &registry_client,
                    const std::shared_ptr<brpc::Server> &server)
        : _service_discoverer(discovery_client),
          _registry_client(registry_client),
          _mysql_client(mysql_client),
          _rpc_server(server) {}

    ~TransmiteServer() {}

    void start()
    {
        _rpc_server->RunUntilAskedToQuit();
    }

private:
    // 服务发现客户端
    Discovery::ptr _service_discoverer;
    // 服务注册客户端
    Registery::ptr _registry_client;
    // mysql客户端
    std::shared_ptr<odb::core::database> _mysql_client;
    std::shared_ptr<brpc::Server> _rpc_server;
};

class TransmiteServerBuild
{
public:
    // 创建mysql对象
    void create_mysql_object(
        const std::string &user,
        const std::string &pswd,
        const std::string &db,
        const std::string &host,
        const std::string &cset,
        int port,
        int connect_pool_count)
    {
        _mysql_client = ODBFactory::create(user, pswd, db, host, cset, port, connect_pool_count);
    }

    void create_discovery_object(const std::string &reg_host,
                                 const std::string &base_service_name,
                                 const std::string &user_service_name)
    {
        _user_service_name = user_service_name;
        _manager_channels = std::make_shared<ServiceManager>();
        // 设置关心的对象
        _manager_channels->declared(_user_service_name);
        auto pub_cb = std::bind(&ServiceManager::onServiceOnline, _manager_channels.get(), std::placeholders::_1, std::placeholders::_2);
        auto del_cb = std::bind(&ServiceManager::onServiceOffline, _manager_channels.get(), std::placeholders::_1, std::placeholders::_2);
        _service_discoverer = std::make_shared<Discovery>(reg_host, base_service_name, pub_cb, del_cb);
    }

    void create_registry_object(const std::string &reg_host, const std::string &service_name, const std::string &access_host)
    {
        _registry_client = std::make_shared<Registery>(reg_host);
        _registry_client->registry(service_name, access_host);
    }

    // 构造rabbitmq客户端对象
    void create_mq_object(const std::string &user,
                        const std::string &passwd,
                        const std::string &host,
                        const std::string &exchange_name,
                        const std::string &queue_name,
                        const std::string &binding_key)
    {
        _routing_key = binding_key;
        _exchange_name = exchange_name;
        _mq_client = std::make_shared<MQClient>(user, passwd, host);
        _mq_client->declareComponents(exchange_name, queue_name, binding_key);
    }

    void create_rpc_server(uint16_t port, int32_t timeout, uint8_t num_threads)
    {
        if (!_mq_client)
        {
            LOG_ERROR("还未初始化mq客户端");
            abort();
        }

        if (!_mysql_client)
        {
            LOG_ERROR("还未初始化mysql客户端");
            abort();
        }

        if (!_manager_channels)
        {
            LOG_ERROR("还未初始化信道管理模块！");
            abort();
        }
        _rpc_server = std::make_shared<brpc::Server>();
        MsgTransmitServiceImpl *transmite_service = new MsgTransmitServiceImpl(
                _user_service_name, _manager_channels, _mysql_client, _exchange_name, _routing_key, _mq_client);
 
        int ret = _rpc_server->AddService(transmite_service,
                                          brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
        if (ret == -1)
        {
            LOG_ERROR("添加Rpc服务失败！");
            abort();
        }

        // 启动服务器
        brpc::ServerOptions options;
        // 连接空闲超时时间，-1表示
        options.idle_timeout_sec = -1;
        // io线程数量
        options.num_threads = 1;
        ret = _rpc_server->Start(port, &options);
        if (ret == -1)
        {
            LOG_ERROR("服务启动失败");
            abort();
        }
    }

    TransmiteServer::ptr build()
    {
        if (!_service_discoverer)
        {
            LOG_ERROR("没有初始化服务发现服务");
            abort();
        }

        if (!_registry_client)
        {
            LOG_ERROR("没有初始化服务注册服务");
            abort();
        }

        if (!_rpc_server)
        {
            LOG_ERROR("没有初始化rpc服务");
            abort();
        }

        TransmiteServer::ptr server = std::make_shared<TransmiteServer>(_mysql_client, _service_discoverer, _registry_client, _rpc_server);

        return server;
    }

private:
    std::string _user_service_name;
    ServiceManager::ptr _manager_channels;  

    std::string _routing_key;
    std::string _exchange_name;
    MQClient::ptr _mq_client;

    // 服务发现客户端
    Discovery::ptr _service_discoverer;
    // 服务注册客户端
    Registery::ptr _registry_client;
    // mysql客户端
    std::shared_ptr<odb::core::database> _mysql_client;
    std::shared_ptr<brpc::Server> _rpc_server;
};
